from datetime import datetime, timedelta, date
from airflow import DAG

from airflow.operators.python import PythonOperator
from airflow.operators.empty import EmptyOperator
import urllib.parse
import psycopg2
import pandas as pd
import sqlalchemy
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker
from airflow.models.xcom import XCom
import time
from random import choice
import pendulum

import requests
from bs4 import BeautifulSoup
import re

from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.common.action_chains import ActionChains
import json
from fake_useragent import UserAgent
from random import choice
import numpy as np
from scipy.stats import norm
# установим seed
np.random.seed(int(round(datetime.now().timestamp())))

LOCAL_TZ = pendulum.timezone("Europe/Moscow")
# ------------------------------------------------------------------------------
# Конфиг с параметрами
# ------------------------------------------------------------------------------
with open('config/config.json') as f:
    config_json = json.load(f)
    conn_string = config_json['db_local_string']
    lot_data_columns = config_json['lot_data_columns']
    search_params = config_json['search_params']
# ------------------------------------------------------------------------------
# user_agents
# ------------------------------------------------------------------------------
with open('config/user_agent_2gis.txt') as file:
    user_agent_lines = file.read().split('\n')
# ------------------------------------------------------------------------------
# SQL
# ------------------------------------------------------------------------------
# сначала берем все объекты, которые нам предстоит отправить (валидные)
q_lots_address = """
select
    lot_id,
    address_formatted,
    square as address_square,
    address_json
from
    (
    select
        ld.lot_id,
        ld.initial_price,
        ld.step,
        ld.application_end_dt,
        ldi.oks_num[1] as oks_num,
        (ldi.square::float[])[1],
        ldi.address_formatted,
        ldi.address_json,
        ldi.longitude,
        ldi.latitude,
        split_part(cord_json->'Point'->>'pos',
        ' ',
        1)::numeric as m_longitude,
        split_part(cord_json->'Point'->>'pos',
        ' ',
        2)::numeric as m_latitude,
        cord_json->>'name' as m_name,
        cord_json->>'description' as m_description,
        (ld.application_end_dt - now()::date) as days_to_period_application_end
    from
        (
        select
            *,
            row_number() over(partition by lot_id
        order by
            id desc) as rn
        from
            etl.lot_data ld
        ) ld
    join (
        select
            *,
            row_number() over(partition by lot_id
        order by
            id desc) as rn
        from
            etl.lot_data_info ld
            where address_formatted is not null) ldi
            using(lot_id)
    where
        ld.rn = 1
        and ldi.rn = 1
        and ld.action_flg <> 'D'
        and ld.trades_section <> 'Аренда/безвозмездное пользование'
        and ld.full_descr not ilike '%%комната%%'
        and is_share = 0
        and address_formatted is not null
        and (ldi.square::float[])[1]>= 20
        and initial_price<20000000
        and ld.lot_id not in (
        select
            lot_id
        from
            etl.lot_messages_to_tg)
        and ld.lot_id not in (
        select
            lot_id
            from etl.lot_data_dom_rf 
        )
            ) t
where
    days_to_period_application_end > 0
    and address_formatted not in ('Россия, Москва, Клинская улица, 2Ак7')
    and (address_json->'metaDataProperty'->'GeocoderMetaData'->'AddressDetails'->'Country'->'AdministrativeArea'->'Locality'->'Thoroughfare'->>'ThoroughfareName') 
    not in ('Береговой проезд')
order by
    days_to_period_application_end asc,
    initial_price asc;
"""

# выбираем proxy
dag_start_hour = datetime.now().hour
proxy_index = dag_start_hour % 3 

# the list of proxy to rotate on 
PROXIES = [
    'http://194.67.219.157:9967',
    'http://194.67.202.42:9169',
    'http://194.28.210.94:9626'
]

random_proxy = PROXIES[proxy_index]



ua = UserAgent()
userAgent = ua.chrome
chrome_options = Options()
chrome_options.add_argument(f'user-agent={userAgent}')
chrome_options.add_argument('--no-sandbox')
chrome_options.add_argument('--window-size=1920,1080')
chrome_options.add_argument('--headless')
chrome_options.add_argument('--disable-dev-shm-usage')
chrome_options.add_argument('--disable-gpu')
chrome_options.add_argument("--disable-notifications")
chrome_options.add_argument("--remote-debugging-port=9222")
chrome_options.add_experimental_option("excludeSwitches", ["enable-automation"])
chrome_options.add_experimental_option('useAutomationExtension', False)
chrome_options.add_experimental_option("excludeSwitches", ["disable-popup-blocking"])
chrome_options.add_argument('--disable-blink-features=AutomationControlled')
chrome_options.add_argument("--disable-extensions")
chrome_options.add_argument(f'--proxy-server={random_proxy}')

# пути к элементам
room_xpath = "//div[@data-id='2']//ul[@class='select__list']/li[1]"
one_flat_xpath = "//div[@data-id='2']//ul[@class='select__list']/li[2]"
two_flat_xpath = "//div[@data-id='2']//ul[@class='select__list']/li[3]"
three_flat_xpath = "//div[@data-id='2']//ul[@class='select__list']/li[4]"

price_per_m_xpath = "//div[@class='bullet__info']/div[1]/span[@class='bullet__data']"
price_lower_border_xpath = "//div[@class='bullet__info']/div[2]/div[@class='bullet__row']/span[1]"
price_upper_border_xpath = "//div[@class='bullet__info']/div[2]/div[@class='bullet__row']/span[2]"
price_xpath = "//div[@class='bullet__info']/div[3]/span[@class='bullet__data']"
station_distance_list_xpath = "//div[@class='bullet__station-container']/div/div[@class='bullet__distance-wr']"
station_name_list_xpath = "//div[@class='bullet__station-container']/div/span[@class='bullet__name-station']"

def send_keys_with_delays(element, value,driver):
    wait = WebDriverWait(driver, 10)
    for i in range(len(value)):
        time.sleep(abs(round(norm(loc = 0.3, scale = 0.1).rvs(), 2)))
        element.send_keys(value[i])
        wait.until(lambda _: element.get_property('value')[:i] == value[:i])

def get_street_house(x):
    elements_list = []
    for element in x['metaDataProperty']['GeocoderMetaData']['Address']['Components']:
        if element['kind'] in ['street', 'house']:
            elements_list.append(element['name'])
    if len(elements_list) == 2:
        return ', '.join(elements_list)
    else:
        return None

def get_dom_rf_data(address_house, address_square):
    station_dict = {}
    url = 'https://xn--h1alcedd.xn--d1aqf.xn--p1ai/services/ocenka-nedvizhimosti/'
    print(url)
    remote_webdriver = 'remote_chromedriver'
    driver =  webdriver.Remote(f'{remote_webdriver}:4444/wd/hub',options=chrome_options)
    driver.maximize_window()
    driver.get(url)
    time.sleep(10)
    # кликаем на окошко с адресом
    address_element = driver.find_element(By.XPATH, "//*[@id='form']/div/div[1]/label/div/input")
    driver.execute_script("arguments[0].click();", address_element)
    driver.save_screenshot('media/img_after_click_address.png')
    # ждем-с
    time.sleep(3)
    # вписываем адрес
    send_keys_with_delays(address_element, 'г. Москва, ' + address_house, driver)
    print('Address was filled in!')
    time.sleep(5)
    # кликаем на окошко с площадью
    square_element = driver.find_element(By.XPATH, "//*[@id='form']/div/div[2]/label/input")
    driver.execute_script("arguments[0].click();", square_element)
    driver.save_screenshot('media/img_after_click_square.png')
    # ждем-с
    time.sleep(3)
    send_keys_with_delays(square_element, str(address_square), driver)
    print('Square was filled in!')
    # кликнем на выпадающий список
    list_element = driver.find_element(By.XPATH, "//*[contains(text(), 'Студия')]")
    # ждем-с
    time.sleep(3)
    driver.execute_script("arguments[0].click();", list_element)
    driver.save_screenshot('media/img_list_element_click.png')
    print('List with flats was clicked!')
    # кликнем на элемент из выпадающего списка
    if address_square < 20:
        path = room_xpath
    elif address_square >= 20 and address_square < 50:
        path = one_flat_xpath
    elif address_square >= 50 and address_square < 80:
        path = two_flat_xpath
    else:
        path = three_flat_xpath
    time.sleep(5)
    flat_element = driver.find_element(By.XPATH, path)
    driver.execute_script("arguments[0].click();", flat_element)
    driver.save_screenshot('media/img_flat_element_click.png')
    print('Flat element was clicked!')
    time.sleep(5)
    button = driver.find_element(By.XPATH, "//button[@class = 'btn full']")
    ActionChains(driver).double_click(button).perform()
    print('All buttons was clicked!')
    driver.save_screenshot('media/img_all_buttons_clicked.png')
    time.sleep(15)
    driver.save_screenshot('media/img_final_result.png')
    # цена за квадратный метр
    price_per_m_element =  driver.find_element(By.XPATH, price_per_m_xpath)
    price_per_m_text = price_per_m_element.text
    print(price_per_m_text)
    price_element = driver.find_element(By.XPATH, price_xpath)
    price_text = price_element.text
    print(price_text)
    price_lower_border_element = driver.find_element(By.XPATH, price_lower_border_xpath)
    price_lower_border_text = price_lower_border_element.text
    print(price_lower_border_text)
    price_upper_border_element = driver.find_element(By.XPATH, price_upper_border_xpath)
    price_upper_border_text = price_upper_border_element.text
    print(price_upper_border_text)
    # станции
    station_distance_list = [element.text for element in driver.find_elements(By.XPATH, station_distance_list_xpath)]
    station_name_list = [element.text for element in driver.find_elements(By.XPATH, station_name_list_xpath)]
    driver.quit()
    for station_distance, station_name in zip(station_distance_list, station_name_list):
        station_dict[station_name] = re.findall(pattern='[\d]+\,?[\d]*', string = station_distance)[0]
    if 'тыс' in price_per_m_text:
        price_per_m = int(float(re.findall(pattern='[\d]+\,?[\d]*', string = price_per_m_text)[0].replace(',', '.'))*1e3)
    if  'млн' in price_text:
        price = int(float(re.findall(pattern='[\d]+\,?[\d]*', string = price_text)[0].replace(',', '.'))*1e6)
    if  'млн' in price_lower_border_text:
        price_lower_border = int(float(re.findall(pattern='[\d]+\,?[\d]*', string = price_lower_border_text)[0].replace(',', '.'))*1e6)
    if  'млн' in price_upper_border_text:
        price_upper_border = int(float(re.findall(pattern='[\d]+\,?[\d]*', string = price_upper_border_text)[0].replace(',', '.'))*1e6)
    
    print('Price per metr: {}'.format(price_per_m))
    print('Price: {}'.format(price_text))
    print('price_lower_borderr: {}'.format(price_lower_border))
    print('price_upper_border: {}'.format(price_upper_border))
    print('Station dict')
    for key in station_dict.keys():
        print('key : ' + key)
        print('value:' + str(station_dict[key]))
    print([price_per_m, price, price_lower_border, price_upper_border, json.dumps(station_dict)])
    return [price_per_m, price, price_lower_border, price_upper_border, json.dumps(station_dict)]
    

def get_address_info():
    conn = create_engine(conn_string)
    df_lots = pd.read_sql(sql = q_lots_address, con = conn).copy()
    df_lots['address_house'] = df_lots.address_json.apply(lambda x: get_street_house(x))
    print('New lots cnt: ' + str(df_lots.shape[0]))
    print('Selected proxy: ' + random_proxy)
    if df_lots.shape[0] == 0:
        raise ValueError('There is no any new lots!')
    for lot_id in df_lots[df_lots['address_house'].notna()].lot_id.unique():
        df = df_lots[df_lots['lot_id']==lot_id].copy()
        print(df.address_house.values[0])
        df[['price_per_m', 'price', 'price_lower_border', 'price_upper_border', 'station_dict']] = df.apply(lambda x: get_dom_rf_data(x['address_house'], x['address_square']), axis = 1, result_type = 'expand')
        df[['lot_id', 'address_house', 'address_square', 'price_per_m', 'price', 'price_lower_border', 'price_upper_border', 'station_dict']].to_sql(name = 'lot_data_dom_rf', schema = 'etl', con = create_engine(conn_string), 
                if_exists = 'append', index = False, dtype={"station_dict": sqlalchemy.types.JSON})
        break


with DAG(
        dag_id='parse_dom_rf',
        start_date = datetime(year = 2023, month = 4, day = 18, hour = 8, minute = 30, tzinfo=LOCAL_TZ),
        schedule_interval='20 0-3,9-23 * * *',
        catchup=False,
        default_args = {
           'owner':'d.kakurin',
           "retries": 1,
           "retry_delay": pendulum.duration(minutes=20),
       },
        tags=['parsing'],
) as dag:
    start_task = EmptyOperator(
        task_id="start_task"
    )

    get_address_info = PythonOperator(
        task_id='get_address_info',
        python_callable=get_address_info
    )

    end_task = EmptyOperator(
        task_id="end_task"
    )


start_task >> get_address_info >> end_task